#!/bin/bash

###############################################################################
# 脚本名称: kafka_offset_check.sh
# 创建时间: 2024-03-25
# 作者: weihr
# 版本: 1.3
#
# 功能描述:
#   批量检查Kafka集群中所有topic的消息偏移量
#   1. 自动获取所有topic列表
#   2. 获取每个topic的分区数
#   3. 获取每个topic的每个分区偏移量
#   4. 将结果保存为CSV格式文件
#
# 使用方法:
#   chmod +x kafka_offset_check.sh
#   ./kafka_offset_check.sh
#
# 输出文件:
#   kafka_offsets_YYYYMMDD_HHMMSS.txt
#
# 注意事项:
#   1. 确保kafka-topics.sh和kafka-run-class.sh在正确的路径下
#   2. 确保有足够的权限访问Kafka集群
#   3. 对于大型集群，检查可能需要较长时间
###############################################################################

# Kafka配置
KAFKA_HOME="/data/kafka"
BROKER_LIST="10.xx.xx.xx:9092,10.xx.xx.xx:9092,10.xx.xx.xx:9092"
ZOOKEEPER_LIST="10.xx.xx.xx:2181,10.xx.xx.xx:2181,10.xx.xx.xx:2181"

# 检查Kafka目录是否存在
if [ ! -d "$KAFKA_HOME" ]; then
    echo "错误: Kafka目录 $KAFKA_HOME 不存在"
    exit 1
fi

# 检查必要的脚本文件
KAFKA_TOPICS="$KAFKA_HOME/bin/kafka-topics.sh"
KAFKA_RUN_CLASS="$KAFKA_HOME/bin/kafka-run-class.sh"

if [ ! -f "$KAFKA_TOPICS" ] || [ ! -f "$KAFKA_RUN_CLASS" ]; then
    echo "错误: Kafka脚本文件不存在"
    echo "请检查以下文件是否存在:"
    echo "- $KAFKA_TOPICS"
    echo "- $KAFKA_RUN_CLASS"
    exit 1
fi

# 创建输出目录
OUTPUT_DIR="kafka_offsets_reports"
mkdir -p "$OUTPUT_DIR"

# 获取所有topic列表
echo "正在获取所有topic列表..."
# 尝试使用 --bootstrap-server
TOPICS=$($KAFKA_TOPICS --bootstrap-server $BROKER_LIST --list 2>/dev/null)

# 如果失败，尝试使用 --zookeeper
if [ -z "$TOPICS" ]; then
    echo "使用 --bootstrap-server 失败，尝试使用 --zookeeper..."
    TOPICS=$($KAFKA_TOPICS --zookeeper $ZOOKEEPER_LIST --list 2>/dev/null)
fi

if [ -z "$TOPICS" ]; then
    echo "错误: 无法获取topic列表，请检查Kafka连接"
    echo "已尝试以下连接方式："
    echo "1. --bootstrap-server $BROKER_LIST"
    echo "2. --zookeeper $ZOOKEEPER_LIST"
    exit 1
fi

# 创建输出文件
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
OUTPUT_FILE="$OUTPUT_DIR/kafka_offsets_$TIMESTAMP.txt"
echo "Topic名称,分区号,分区偏移量,最后更新时间" > $OUTPUT_FILE

# 记录开始时间
START_TIME=$(date +%s)

# 遍历每个topic
TOTAL_TOPICS=$(echo "$TOPICS" | wc -l)
CURRENT=0

for topic in $TOPICS; do
    CURRENT=$((CURRENT + 1))
    echo "[$CURRENT/$TOTAL_TOPICS] 正在处理 topic: $topic"
    
    # 获取topic的分区数
    PARTITIONS=$($KAFKA_TOPICS --zookeeper $ZOOKEEPER_LIST --describe --topic $topic 2>/dev/null | grep "PartitionCount" | awk '{print $2}')
    
    # 获取每个分区的偏移量
    $KAFKA_RUN_CLASS kafka.tools.GetOffsetShell \
        --broker-list $BROKER_LIST \
        --topic $topic \
        --time -1 | while IFS=":" read -r topic_name partition offset; do
            # 获取最后更新时间
            LAST_UPDATE=$(date "+%Y-%m-%d %H:%M:%S")
            # 输出结果
            echo "$topic_name,$partition,$offset,$LAST_UPDATE" >> $OUTPUT_FILE
        done
done

# 计算执行时间
END_TIME=$(date +%s)
DURATION=$((END_TIME - START_TIME))

echo "检查完成！"
echo "总耗时: ${DURATION} 秒"
echo "结果已保存到: $OUTPUT_FILE"

# 显示结果
echo -e "\n结果预览:"
column -t -s ',' $OUTPUT_FILE